airbyte_cdk/sources/file_based/exceptions.py (1)

132-149: Add type annotations to improve type safety, wdyt?

The message formatting looks great! Consider adding type annotations:

-def format_duplicate_files_error_message(stream_name: str, duplicated_files_names: List):
+def format_duplicate_files_error_message(stream_name: str, duplicated_files_names: List[Dict[str, List[str]]]) -> str:

🧰 Tools

🪛 GitHub Actions: Linters

[error] 132-132: Function is missing a return type annotation

---

[error] 132-132: Missing type parameters for generic type "List"

airbyte_cdk/sources/file_based/file_based_stream_reader.py (1)

138-146: Consider improving null safety in the config check, wdyt?

The logic looks good, but we could make the null checks more explicit:

     def preserve_subdirectories_directories(self) -> bool:
         # fall back to preserve subdirectories if config is not present or incomplete
         if (
             self.config
             and hasattr(self.config, "delivery_options")
-            and hasattr(self.config.delivery_options, "preserve_subdirectories_directories")
+            and self.config.delivery_options is not None
+            and hasattr(self.config.delivery_options, "preserve_subdirectories_directories")
         ):
             return self.config.delivery_options.preserve_subdirectories_directories
         return True

🧰 Tools

🪛 GitHub Actions: Linters

[error] 145-145: Item "None" of "DeliveryOptions | None" has no attribute "preserve_subdirectories_directories"

airbyte_cdk/sources/file_based/config/abstract_file_based_spec.py (1)

17-22: Consider enhancing the field description for clarity, wdyt?

The implementation looks good! Consider making the description more descriptive:

     preserve_subdirectories_directories: bool = Field(
         True,
-        description="Flag indicating we should preserve subdirectories directories",
+        description="When enabled, preserves the subdirectory structure of files during transfer. When disabled, all files are stored in the root directory.",
     )

airbyte_cdk/sources/file_based/file_based_source.py (1)

392-399: Add return type and improve null safety, wdyt?

The logic looks good, but we could improve type safety:

     @staticmethod
-    def _preserve_subdirectories_directories(parsed_config: AbstractFileBasedSpec):
+    def _preserve_subdirectories_directories(parsed_config: AbstractFileBasedSpec) -> bool:
         # fall back to preserve subdirectories if config is not present or incomplete
         if hasattr(parsed_config, "delivery_options") and hasattr(
-            parsed_config.delivery_options, "preserve_subdirectories_directories"
+            parsed_config.delivery_options, "preserve_subdirectories_directories"
+        ) and parsed_config.delivery_options is not None:
         ):
             return parsed_config.delivery_options.preserve_subdirectories_directories
         return True

🧰 Tools

🪛 GitHub Actions: Linters

[error] 393-393: Function is missing a return type annotation

---

[error] 398-398: Item "None" of "DeliveryOptions | None" has no attribute "preserve_subdirectories_directories"

airbyte_cdk/sources/file_based/stream/default_file_based_stream.py (2)

50-51: Consider a more concise property name.

The property name preserve_subdirectories_directories seems redundant with "directories". What do you think about shortening it to just preserve_subdirectories? wdyt?

-    preserve_subdirectories_directories = True
+    preserve_subdirectories = True

Also applies to: 59-59

---

64-66: Update initialization if property name changes.

If you agree with shortening the property name, we should update the initialization too. wdyt?

-            self.preserve_subdirectories_directories = kwargs.pop(
-                self.PRESERVE_SUBDIRECTORIES_KW, True
-            )
+            self.preserve_subdirectories = kwargs.pop(self.PRESERVE_SUBDIRECTORIES_KW, True)

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/file_based/stream/default_file_based_stream.py (3)

59-66: Consider simplifying the property name.

The property name preserve_subdirectories_directories seems to have a redundant "directories" suffix. Would preserve_subdirectories be clearer and more concise? The initialization logic looks good though! Wdyt?

-    preserve_subdirectories_directories = True
+    preserve_subdirectories = True

     def __init__(self, **kwargs: Any):
         if self.FILE_TRANSFER_KW in kwargs:
             self.use_file_transfer = kwargs.pop(self.FILE_TRANSFER_KW, False)
-            self.preserve_subdirectories_directories = kwargs.pop(
+            self.preserve_subdirectories = kwargs.pop(
                 self.PRESERVE_SUBDIRECTORIES_KW, True
             )

---

111-126: Add type annotations and consider renaming the method.

The duplicate detection logic looks solid! A few suggestions to make it even better:

1. The method name could be more Pythonic. Maybe _get_duplicated_filenames?
2. Let's add proper type annotations to fix the pipeline failures.

-    def _duplicated_files_names(self, slices: List) -> list[dict]:
+    def _get_duplicated_filenames(
+        self,
+        slices: List[Dict[str, List[RemoteFile]]]
+    ) -> List[Dict[str, List[str]]]:

🧰 Tools

🪛 GitHub Actions: Linters

[error] 111-111: Missing type parameters for generic type "List"

---

[error] 111-111: Missing type parameters for generic type "dict"

---

133-145: Consider extracting the duplicate check into a guard clause.

The logic looks correct, but we could make it more readable by extracting the duplicate check into a guard clause at the beginning of the method. Wdyt?

     def compute_slices(self) -> Iterable[Optional[Mapping[str, Any]]]:
         # Sort files by last_modified, uri and return them grouped by last_modified
         all_files = self.list_files()
         files_to_read = self._cursor.get_files_to_sync(all_files, self.logger)
         sorted_files_to_read = sorted(files_to_read, key=lambda f: (f.last_modified, f.uri))
         slices = [
             {self.FILES_KEY: list(group[1])}
             for group in itertools.groupby(sorted_files_to_read, lambda f: f.last_modified)
         ]
-        if slices and not self.preserve_subdirectories_directories:
-            duplicated_files_names = self._duplicated_files_names(slices)
-            if duplicated_files_names:
-                raise DuplicatedFilesError(
-                    format_duplicate_files_error_message(
-                        stream_name=self.name, duplicated_files_names=duplicated_files_names
-                    ),
-                    stream=self.name,
-                )
+        if slices and not self.preserve_subdirectories_directories:
+            self._check_for_duplicates(slices)
         return slices

+    def _check_for_duplicates(self, slices: List[Dict[str, List[RemoteFile]]]) -> None:
+        duplicated_files_names = self._get_duplicated_filenames(slices)
+        if duplicated_files_names:
+            raise DuplicatedFilesError(
+                format_duplicate_files_error_message(
+                    stream_name=self.name,
+                    duplicated_files_names=duplicated_files_names
+                ),
+                stream=self.name,
+            )

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/file_based/file_based_source.py (1)

392-401: Well-structured configuration handling.

The method follows good practices:

• Safely handles missing or incomplete configuration
• Preserves backward compatibility by defaulting to True
• Clear and descriptive attribute checks

What do you think about adding a docstring to explain the default behavior? wdyt?

 @staticmethod
 def _preserve_subdirectories_directories(parsed_config: AbstractFileBasedSpec) -> bool:
+    """
+    Determine whether to preserve subdirectories based on the configuration.
+    
+    Returns True if:
+    - The configuration is missing
+    - The delivery_options is not set
+    - The preserve_subdirectories_directories flag is not set
+    """
     # fall back to preserve subdirectories if config is not present or incomplete

airbyte_cdk/sources/file_based/stream/default_file_based_stream.py (2)

111-128: Well-implemented duplicate detection.

The implementation is efficient using sets and defaultdict. A few suggestions to make it even better:

1. Would you consider adding type hints for the return value's inner types? wdyt?
2. How about adding a docstring to explain the method's purpose and return format?

 def _duplicated_files_names(
     self, slices: List[dict[str, List[RemoteFile]]]
-) -> List[dict[str, List[str]]]:
+) -> List[dict[str, list[str]]]:
+    """
+    Identify duplicate file names across all slices.
+    
+    Args:
+        slices: List of slices containing RemoteFile objects
+        
+    Returns:
+        List of dictionaries mapping duplicate file names to their full paths
+    """

---

139-147: Clean integration of duplicate detection.

The duplicate check is well-integrated into the existing slice computation logic. However, consider extracting the duplicate check into a separate method for better readability. wdyt?

 def compute_slices(self) -> Iterable[Optional[Mapping[str, Any]]]:
+    def _check_for_duplicates(slices: List[dict[str, List[RemoteFile]]]) -> None:
+        if not self.preserve_subdirectories_directories:
+            duplicated_files_names = self._duplicated_files_names(slices)
+            if duplicated_files_names:
+                raise DuplicatedFilesError(
+                    format_duplicate_files_error_message(
+                        stream_name=self.name, duplicated_files_names=duplicated_files_names
+                    ),
+                    stream=self.name,
+                )
+
     all_files = self.list_files()
     files_to_read = self._cursor.get_files_to_sync(all_files, self.logger)
     sorted_files_to_read = sorted(files_to_read, key=lambda f: (f.last_modified, f.uri))
     slices = [
         {self.FILES_KEY: list(group[1])}
         for group in itertools.groupby(sorted_files_to_read, lambda f: f.last_modified)
     ]
-    if slices and not self.preserve_subdirectories_directories:
-        duplicated_files_names = self._duplicated_files_names(slices)
-        if duplicated_files_names:
-            raise DuplicatedFilesError(
-                format_duplicate_files_error_message(
-                    stream_name=self.name, duplicated_files_names=duplicated_files_names
-                ),
-                stream=self.name,
-            )
+    if slices:
+        _check_for_duplicates(slices)
     return slices

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/file_based/file_based_stream_reader.py (1)

Line range hint 172-182: The logic looks good! Would you consider some readability improvements?

The implementation correctly uses the new configuration option, but we could make it even clearer with:

1. More descriptive variable names (e.g., relative_file_path instead of file_relative_path)
2. A more detailed comment explaining the path transformation logic

What do you think about this diff?

-    # Remove left slashes from source path format to make relative path for writing locally
-    file_relative_path = file.uri.lstrip("/")
+    # Transform the source file path into a relative path for local writing:
+    # - If preserving directories: maintain the path structure but remove leading slashes
+    # - If not preserving: use only the filename
+    relative_file_path = file.uri.lstrip("/") if preserve_subdirectories_directories else path.basename(file.uri)

airbyte_cdk/sources/file_based/config/abstract_file_based_spec.py (2)

34-38: The field definition looks good! Should we enhance the description?

The implementation is solid, but the description could be more informative about what this setting actually does. What do you think about making it clearer for users?

     preserve_subdirectories_directories: bool = Field(
         title="Preserve Subdirectories Directories",
-        description="Flag indicating we should preserve subdirectories directories",
+        description="When enabled, maintains the original directory structure of files when copying them to the destination. When disabled, all files are copied to a flat structure in the destination directory.",
         default=True,
     )

---

74-78: Should we use the same enhanced description here for consistency?

     preserve_subdirectories_directories: bool = Field(
         title="Preserve Subdirectories Directories",
-        description="Flag indicating we should preserve subdirectories directories",
+        description="When enabled, maintains the original directory structure of files when copying them to the destination. When disabled, all files are copied to a flat structure in the destination directory.",
         default=True,
     )

unit_tests/sources/file_based/scenarios/csv_scenarios.py (1)

526-544: The test configuration looks good! Should we add more test coverage?

The implementation correctly includes the new field in the test scenarios. However, we might want to add test cases that specifically verify the behavior when:

1. preserve_subdirectories_directories is set to False
2. Files are in nested subdirectories
3. Edge cases like empty directories or files with identical names in different subdirectories

Would you like me to help draft these additional test scenarios?

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/file_based/file_based_source.py (1)

393-393: What do you think about simplifying the method name? 🤔

The current name _preserve_subdirectories_directories seems a bit redundant with "directories" appearing twice. Maybe we could simplify it to _preserve_subdirectories? wdyt?

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

unit_tests/sources/file_based/test_file_based_stream_reader.py (2)

Line range hint 111-128: LGTM! Consider a small optimization for memory efficiency

The implementation looks clean and efficient. Would you consider using a generator expression instead of a list comprehension for the return value to optimize memory usage when dealing with large numbers of duplicates? wdyt?

-        return [
+        return (
             {duplicated_file: file_paths[duplicated_file]}
             for duplicated_file in duplicates_file_names
-        ]
+        )

---

371-440: LGTM! Consider adding edge cases

The test coverage looks good! Would you consider adding a few edge cases to make it even more robust? For example:

• Empty directory paths
• Paths with special characters
• Very long paths
• Paths with spaces

This would help ensure the feature handles all scenarios gracefully. wdyt?

airbyte_cdk/sources/file_based/file_based_source.py (1)

392-401: LGTM! Consider enhancing the comment

The implementation looks solid! Would you consider adding a more descriptive docstring to explain:

• The purpose of the method
• The expected config structure
• The default behavior and why it defaults to True
• Example usage

Something like:

@staticmethod
def _preserve_directory_structure(parsed_config: AbstractFileBasedSpec) -> bool:
    """
    Determines whether to preserve directory structure during file transfer.
    
    When enabled, files maintain their subdirectory paths in the destination.
    When disabled, files are flattened to the root of the destination.
    
    Args:
        parsed_config: The parsed configuration containing delivery method settings
        
    Returns:
        True if directory structure should be preserved (default), False otherwise
    """

wdyt?

🧰 Tools

🪛 GitHub Actions: Linters

[warning] File requires formatting. Run 'ruff format' to fix code formatting issues.

airbyte_cdk/sources/file_based/stream/default_file_based_stream.py (1)

111-128: Consider using a more memory-efficient approach for large file sets, wdyt?

The current implementation stores all file paths in memory. For very large file sets, this could be memory intensive. Consider using a generator-based approach or processing files in chunks if memory usage becomes a concern.

Here's a potential optimization:

    def _duplicated_files_names(
        self, slices: List[dict[str, List[RemoteFile]]]
    ) -> List[dict[str, List[str]]]:
-       seen_file_names = set()
-       duplicates_file_names = set()
-       file_paths = defaultdict(list)
+       seen_file_names: Dict[str, List[str]] = defaultdict(list)
        for file_slice in slices:
            for file_found in file_slice[self.FILES_KEY]:
                file_name = path.basename(file_found.uri)
-               if file_name not in seen_file_names:
-                   seen_file_names.add(file_name)
-               else:
-                   duplicates_file_names.add(file_name)
-               file_paths[file_name].append(file_found.uri)
+               seen_file_names[file_name].append(file_found.uri)
        return [
-           {duplicated_file: file_paths[duplicated_file]}
-           for duplicated_file in duplicates_file_names
+           {file_name: paths}
+           for file_name, paths in seen_file_names.items()
+           if len(paths) > 1
        ]

unit_tests/sources/file_based/stream/test_default_file_based_stream.py (1)

370-469: Comprehensive test coverage for the new functionality!

The test class covers both successful and error scenarios with realistic test data. The error message validation is thorough.

Consider adding a test case for empty slices, wdyt?

def test_when_compute_slices_with_empty_slices(self) -> None:
    with (
        mock.patch.object(DefaultFileBasedStream, "list_files", return_value=[]),
        mock.patch.object(self._stream._cursor, "get_files_to_sync", return_value=[]),
    ):
        returned_slices = self._stream.compute_slices()
    assert returned_slices == []

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/file_based/file_based_source.py (2)

301-301: Consider adding return type hints to helper methods?

The helper methods _use_file_transfer and _preserve_directory_structure could benefit from explicit return type hints, wdyt? This would make the API contract clearer:

    @staticmethod
-   def _use_file_transfer(parsed_config: AbstractFileBasedSpec) -> bool:
+   def _use_file_transfer(parsed_config: AbstractFileBasedSpec) -> bool:

Also applies to: 313-314

---

390-399: Want to make this more concise using the walrus operator?

The implementation could be simplified while maintaining the same logic, wdyt?

    @staticmethod
    def _preserve_directory_structure(parsed_config: AbstractFileBasedSpec) -> bool:
-       # fall back to preserve subdirectories if config is not present or incomplete
-       if (
-           FileBasedSource._use_file_transfer(parsed_config)
-           and hasattr(parsed_config.delivery_method, "preserve_directory_structure")
-           and parsed_config.delivery_method.preserve_directory_structure is not None
-       ):
-           return parsed_config.delivery_method.preserve_directory_structure
-       return True
+       # fall back to preserve subdirectories if config is not present or incomplete
+       return (preserve := getattr(parsed_config.delivery_method, "preserve_directory_structure", None)) is not None \
+           and FileBasedSource._use_file_transfer(parsed_config) \
+           and preserve

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/file_based/file_based_source.py (1)

Test files need parameter updates for DefaultFileBasedStream constructor

Some test instances are missing the new parameters. Would you mind updating these?

• test_default_file_based_stream.py:100: Missing both use_file_transfer and preserve_directory_structure
• test_default_file_based_stream.py:297: Missing preserve_directory_structure
• test_adapters.py:225: Missing both parameters

The production code looks good with both parameters properly handled! 🎉

🔗 Analysis chain

Line range hint 301-314: The parameter changes look good! Quick verification needed.

The changes to _make_default_stream look well-structured. The method now computes use_file_transfer internally and adds support for preserve_directory_structure. Would you mind verifying that all callers of DefaultFileBasedStream constructor are updated to handle these new parameters?

🏁 Scripts executed

The following scripts were executed for the analysis:

Script:

#!/bin/bash
# Description: Find all instantiations of DefaultFileBasedStream to verify constructor parameter updates
ast-grep --pattern 'DefaultFileBasedStream($$$)'

Length of output: 8901

airbyte_cdk/sources/file_based/stream/default_file_based_stream.py (1)

111-121: Consider using a more descriptive return type hint, wdyt?

The method implementation is clean and efficient. However, the return type could be more descriptive to better document the structure:

    def _duplicated_files_names(
-        self, slices: List[dict[str, List[RemoteFile]]]
-    ) -> List[dict[str, List[str]]]:
+        self, slices: List[dict[str, List[RemoteFile]]]
+    ) -> List[dict[str, List[str]]]:  # Dict maps duplicate filename to list of full paths

airbyte_cdk/sources/file_based/file_based_source.py (1)

390-410: Great implementation! Consider a more concise version?

The method is well-documented and handles edge cases correctly. What do you think about making it more concise like this?

     @staticmethod
     def _preserve_directory_structure(parsed_config: AbstractFileBasedSpec) -> bool:
         """
         Determines whether to preserve directory structure during file transfer.
 
         When enabled, files maintain their subdirectory paths in the destination.
         When disabled, files are flattened to the root of the destination.
 
         Args:
             parsed_config: The parsed configuration containing delivery method settings
 
         Returns:
             True if directory structure should be preserved (default), False otherwise
         """
-        if (
-            FileBasedSource._use_file_transfer(parsed_config)
-            and hasattr(parsed_config.delivery_method, "preserve_directory_structure")
-            and parsed_config.delivery_method.preserve_directory_structure is not None
-        ):
-            return parsed_config.delivery_method.preserve_directory_structure
-        return True
+        return (not FileBasedSource._use_file_transfer(parsed_config) or
+                not hasattr(parsed_config.delivery_method, "preserve_directory_structure") or
+                parsed_config.delivery_method.preserve_directory_structure is None or
+                parsed_config.delivery_method.preserve_directory_structure)

The suggested version:

1. Uses boolean logic to combine conditions
2. Maintains the same behavior
3. Reduces nesting
4. Makes the default True case more explicit

What do you think? 🤔

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/file_based/file_types/unstructured_parser.py (2)

52-58: Consider improving error handling.

The error handling could be more specific. Instead of catching all OSErrors, would it make sense to catch specific exceptions like PermissionError and provide more detailed error messages? For example:

     try:
         nltk_data_dir = "/airbyte/nltk_data"
         os.makedirs(nltk_data_dir, exist_ok=True)
-    except OSError:
+    except PermissionError:
         nltk_data_dir = "/tmp/nltk_data"
-        os.makedirs(nltk_data_dir, exist_ok=True)
+        try:
+            os.makedirs(nltk_data_dir, exist_ok=True)
+        except PermissionError as e:
+            raise RuntimeError(f"Unable to create NLTK data directory: {e}")

---

62-70: Consider consolidating NLTK downloads.

The current implementation downloads tokenizers individually. Would it be cleaner to consolidate these into a list? For example:

-    nltk.download("punkt", download_dir=nltk_data_dir)
-    nltk.download("punkt_tab", download_dir=nltk_data_dir)
-    nltk.download("averaged_perceptron_tagger_eng", download_dir=nltk_data_dir)
+    tokenizers = ["punkt", "punkt_tab", "averaged_perceptron_tagger_eng"]
+    for tokenizer in tokenizers:
+        nltk.download(tokenizer, download_dir=nltk_data_dir)

This would make it easier to maintain the list of required tokenizers. wdyt?

🧰 Tools

🪛 GitHub Actions: Linters

[error] 62-62: Call to untyped function "get_ntlk_temp_folder" in typed context

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/file_based/file_types/unstructured_parser.py (1)

62-70: Add logging for better observability.

Would it help to add logging to track download progress and failures? Here's a suggestion:

 try:
+    logger = logging.getLogger(__name__)
+    logger.info(f"Initializing NLTK data directory at {nltk_data_dir}")
     nltk.data.path.append(nltk_data_dir)
     nltk.data.find("tokenizers/punkt.zip")
     nltk.data.find("tokenizers/punkt_tab.zip")
     nltk.data.find("tokenizers/averaged_perceptron_tagger_eng.zip")
 except LookupError:
+    logger.info("Downloading required NLTK tokenizers...")
     nltk.download("punkt", download_dir=nltk_data_dir)
     nltk.download("punkt_tab", download_dir=nltk_data_dir)
     nltk.download("averaged_perceptron_tagger_eng", download_dir=nltk_data_dir)
+    logger.info("Successfully downloaded all required NLTK tokenizers")

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/file_based/file_types/unstructured_parser.py (2)

47-59: Fix typo in function name and enhance error handling.

The implementation looks good, but I noticed a few things we could improve. What do you think about:

1. Fixing the typo in the function name (ntlk → nltk)?
2. Adding an error message in the except block to help with debugging?
3. Using constants for the paths?

+AIRBYTE_NLTK_DATA_DIR = "/airbyte/nltk_data"
+TMP_NLTK_DATA_DIR = "/tmp/nltk_data"

-def get_ntlk_temp_folder() -> str:
+def get_nltk_temp_folder() -> str:
     """
     For non-root connectors /tmp is not currently writable, but we should allow it in the future.
     It's safe to use /airbyte for now. Fallback to /tmp for local development.
     """
     try:
-        nltk_data_dir = "/airbyte/nltk_data"
+        nltk_data_dir = AIRBYTE_NLTK_DATA_DIR
         os.makedirs(nltk_data_dir, exist_ok=True)
-    except OSError:
-        nltk_data_dir = "/tmp/nltk_data"
+    except OSError as e:
+        logger.debug(f"Failed to create {AIRBYTE_NLTK_DATA_DIR}: {e}. Falling back to {TMP_NLTK_DATA_DIR}")
+        nltk_data_dir = TMP_NLTK_DATA_DIR
         os.makedirs(nltk_data_dir, exist_ok=True)
     return nltk_data_dir

---

62-70: Consider more granular error handling and logging.

The NLTK data initialization looks good, but what do you think about making it more robust and informative? Here's a suggestion:

 try:
     nltk_data_dir = get_nltk_temp_folder()
     nltk.data.path.append(nltk_data_dir)
+    # Try loading each tokenizer separately for better error handling
     nltk.data.find("tokenizers/punkt.zip")
+except LookupError:
+    logger.info("Downloading punkt tokenizer...")
+    nltk.download("punkt", download_dir=nltk_data_dir, quiet=True)
+
+try:
     nltk.data.find("tokenizers/punkt_tab.zip")
+except LookupError:
+    logger.info("Downloading punkt_tab tokenizer...")
+    nltk.download("punkt_tab", download_dir=nltk_data_dir, quiet=True)
+
+try:
     nltk.data.find("tokenizers/averaged_perceptron_tagger_eng.zip")
-except LookupError:
-    nltk.download("punkt", download_dir=nltk_data_dir, quiet=True)
-    nltk.download("punkt_tab", download_dir=nltk_data_dir, quiet=True)
-    nltk.download("averaged_perceptron_tagger_eng", download_dir=nltk_data_dir, quiet=True)
+except LookupError:
+    logger.info("Downloading averaged_perceptron_tagger_eng tokenizer...")
+    nltk.download("averaged_perceptron_tagger_eng", download_dir=nltk_data_dir, quiet=True)

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

airbyte_cdk/sources/file_based/file_types/unstructured_parser.py (1)

46-47: Consider using environment variables for NLTK data paths?

The paths are currently hardcoded. For better flexibility across different environments, what do you think about using environment variables with fallbacks? Something like:

-AIRBYTE_NLTK_DATA_DIR = "/airbyte/nltk_data"
-TMP_NLTK_DATA_DIR = "/tmp/nltk_data"
+AIRBYTE_NLTK_DATA_DIR = os.getenv("AIRBYTE_NLTK_DATA_DIR", "/airbyte/nltk_data")
+TMP_NLTK_DATA_DIR = os.getenv("TMP_NLTK_DATA_DIR", "/tmp/nltk_data")

This would make it easier to configure paths in different deployment scenarios, wdyt?

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro